package com.lineengine.lineengine.pipeline;


import com.lineengine.lineengine.common.ErrorType;
import com.lineengine.lineengine.context.IContext;
import com.lineengine.lineengine.common.CommonErrorCode;
import com.lineengine.lineengine.node.INode;
import com.lineengine.lineengine.result.DefaultResult;
import com.lineengine.lineengine.result.IResult;
import com.lineengine.lineengine.result.ParallelResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.task.AsyncTaskExecutor;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import static java.util.stream.Collectors.toList;

/**
 * <p>
 * 并发Pipeline
 * </p>
 *
 * @author wenrong.yu
 * @since 2020-11-08 09:58
 */
public class ParallelPipeline<T extends IContext<T>> extends AbstractPipeline<T> {
    private static final Logger LOGGER = LoggerFactory.getLogger(ParallelPipeline.class);

    private static final int TIME_OUT = 5;

    private AsyncTaskExecutor asyncTaskExecutor;

    ParallelPipeline() {
    }

    @Override
    protected void preCheck() {
        super.preCheck();
        if (this.asyncTaskExecutor == null) {

        }
    }

    @Override
    public IResult<T> doExecute(T context) {
        try {
            // 并发执行
            List<Future<IResult<T>>> futures = parallelExecute(context);
            // 获取结果
            List<IResult<T>> results = extractResults(futures);
            return new ParallelResult<>(results);
        } catch (Exception e) {
            LOGGER.warn("Occurs Exception. Failed to execute parallel pipeline.", e);

            return DefaultResult.fail(ErrorType.SERVER, CommonErrorCode.INTERNAL_SERVER_ERROR);
        }
    }

    /**
     * 并发执行
     *
     * @param context 上下文
     * @return 业务并发future
     */
    private List<Future<IResult<T>>> parallelExecute(T context) {
        return this.nodes.stream().map(node -> this.asyncTaskExecutor.submit(() -> node.execute(context))).collect(toList());
    }

    /**
     * 获取各个并发业务执行结果，默认等待30秒，超时抛出EngineRuntimeException
     *
     * @param futures 业务并发future
     * @return 业务执行结果集合
     */
    private List<IResult<T>> extractResults(List<Future<IResult<T>>> futures) throws ExecutionException, InterruptedException, TimeoutException {
        List<IResult<T>> results = new ArrayList<>(futures.size());
        for (Future<IResult<T>> future : futures) {
            results.add(future.get(TIME_OUT, TimeUnit.SECONDS));
        }
        return results;
    }

    /**
     * <p>
     * 建造器
     * </p>
     *
     * @author wenrong.yu
     * @since 2020-11-08 09:55
     */
    public static class Builder<T extends IContext<T>> {
        private final List<INode<T>> nodes = new ArrayList<>();
        private AsyncTaskExecutor asyncTaskExecutor;

        public Builder<T> add(INode<T> node) {
            this.nodes.add(node);
            return this;
        }

        public Builder<T> add(List<INode<T>> nodes) {
            this.nodes.addAll(nodes);
            return this;
        }

        public Builder<T> parallel(AsyncTaskExecutor asyncTaskExecutor) {
            this.asyncTaskExecutor = asyncTaskExecutor;
            return this;
        }

        public ParallelPipeline<T> build() {
            ParallelPipeline<T> parallelPipeline = new ParallelPipeline<>();
            parallelPipeline.nodes = this.nodes;
            parallelPipeline.asyncTaskExecutor = this.asyncTaskExecutor;
            return parallelPipeline;
        }
    }
}
